package com.ebs.md.container.esper;

/**
 * Perforce Revision:
 *  $File: //depot/MarketData/fxmonitor/main/src/main/java/com/ebs/md/container/esper/AbsStatement.java $
 *  $Revision: #5 $  $Change: 114924 $  $DateTime: 2012/09/13 09:11:55 $  $Author: Aviad.Yosef $
 */

import java.util.List;

import org.apache.log4j.Logger;
import org.springframework.util.Assert;

import com.ebs.md.fxmonitor.marketmonitor.config.INameableConfig;
import com.espertech.esper.client.EPAdministrator;
import com.espertech.esper.client.EPServiceProvider;
import com.espertech.esper.client.EPStatement;
import com.espertech.esper.client.EventBean;
import com.espertech.esper.client.StatementAwareUpdateListener;

/**
 * AbsStatement
 * 
 * @author asher
 */
public abstract class AbsStatement<CONF extends INameableConfig>
implements  StatementAwareUpdateListener
{
	final protected Logger logger = Logger.getLogger(getClass());

	protected String               eplExpressionText;
	protected List<StatementAwareUpdateListener> listeners;
	protected boolean              configRequired = true;
	protected CONF                 config;
	protected String               region;
	protected String               statementName = getClass().getSimpleName();
	
	// managed locally
	protected EPStatement          epStatement;
	
	
	public void init() {
		Assert.hasText(eplExpressionText,  "'eplExpressionText' should have text");
//		Assert.hasText(statementName,      "'statementName' should have text");
		
		if ( configRequired ) {
			Assert.notNull(config, "'config' should not be null");
		}
		Assert.hasText(region,     "'region' should have text");
	}
	
	public void setEplExpressionText(String eplExpressionText) {
		this.eplExpressionText = eplExpressionText;
	}

	public void setListeners(List<StatementAwareUpdateListener> listeners) {
		this.listeners = listeners;
	}

	public void setConfig(CONF config) {
		this.config = config;
	}

	public void setConfigRequired(boolean configRequired) {
		this.configRequired = configRequired;
	}

	public void setRegion(String region) {
		this.region = region;
	}

	public void setStatementName(String statementName) {
		this.statementName = statementName;
	}

	public String getRegion() {
		return region;
	}

	abstract protected String createEplExpression()
	;

	@Override // implement
	public void update(EventBean[] newEvents, EventBean[] oldEvents, EPStatement epstatement, EPServiceProvider epserviceprovider) {
		// by default, do nothing...
	}
	
	public void prepare(EPAdministrator epAdministrator) {
		String eplExpr = createEplExpression();
		logger.info("epStatement" + eplExpr);
		
		destrotyOlderStatement();
		
		epStatement = epAdministrator.createEPL(eplExpr, statementName);

		registerListeners();
	}
	
	protected void destrotyOlderStatement() {
		if ( epStatement != null ) {
			epStatement.destroy();
		}
		epStatement = null;
	}
	
	/**
	 * call method, only after:
	 *  1) listeners are set
	 *  2) epStatement is set
	 * 
	 */
	protected void registerListeners() {
		epStatement.addListener(this);
		
		if ( listeners != null ) {
			for ( StatementAwareUpdateListener curr : listeners ) {
				epStatement.addListener(curr);
			}
		}
	}
	
}
